#include <cmath>
#include <string>
#include <fstream>
#include <iostream>
#include <sstream>
#include <vector>
#include <map>
#include <set>
#include "def.h"

using namespace std;

#include "Mutex.h"
#include "Cond.h"
#include "BiContainer.h"
#include "Lockable.h"
#include "VarCol.h"
#include "Formattable.h"
#include "TableState.h"	
#include "Table.h"

Table::Table (const string& table_name) : Formattable (table_name) {
 readFormat();
}

bool Table::open (TableState& state) {

 if (state.page > numPages - 1) {
  return false;
 }

 pages[state.page].open ( state );

 return true;
}

bool Table::position (TableState& state, int rowId) {
 int position = rowId  * rowLength;
}

bool Table::getSortedIndexList (vector<string>& columnNames, vector< BiContainer<int> >& indexList) {
 BiContainer<int> container;

 vector<string>::iterator colNameItrBeg = columnNames.begin();
 vector<string>::iterator colNameItr = columnNames.begin();
 vector<string>::iterator colNameItrEnd = columnNames.end();

 while (colNameItr < colNameItrEnd) {
  int index = getColumnIndex ( *colNameItr );
  
  if (index == -1) {
   return false;
  }
  else {
   container.set (index, colNameItr - colNameItrBeg);
   indexList.push_back ( container );
  }  
  
  colNameItr++;
 }

 sort (indexList.begin(), indexList.end());
 return true;
}

bool Table::getRow (TableState& state, vector<string>& rowColumns) {

 if (state.data.peek() == EOF) { /* if at end of current page
                                    open the next one */
  state.page += 1;
  state.data.close ();
  if ( !open ( state ) ) return false;
 }
 
 if (state.columnNames.size() == 0) { /* read row in table's own order */
  int i = 0;
  int curLen = rowColumns.size ();
  rowColumns.resize( curLen + numColumns );

  pages[ state.page ].attachReader();

  while (i < numColumns) {

   rowColumns[ curLen + i ].resize ( colLen[i] );  
   state.data.read ( &rowColumns[ curLen + i ][0], colLen[i]);

   if ( isVarCol (i) ) { /* This column is variable-length, use corresponding VarCol instance */
    varCols[i].get (*state.varColStreams[i], atoi ( rowColumns [ curLen + i ].c_str() ), rowColumns [ curLen + i ]);
   }
  
   i++;
  }

  pages[ state.page ].detachReader();
 }
 else { /* read row in a custom order provided by state */

  if (state.sortedIndexList.size () == 0) {
   getSortedIndexList (state.columnNames, state.sortedIndexList);
  }

  int len = state.sortedIndexList.size();

  int curLen = rowColumns.size ();
  rowColumns.resize ( curLen + len ); /* Return as many columns as requested */

  int i = 0;
  int j = 0;
 
  pages[ state.page ].attachReader ();

  while (i < numColumns && j < len) { /* Loop over all columns, not just the requested (until all the requested are filled) */  
   if (i == state.sortedIndexList[j].key) { /* if the column is requested: seek and take input */
    rowColumns[ curLen + state.sortedIndexList[j].value ].resize ( colLen[i] );
    state.data.read( &rowColumns[ curLen + state.sortedIndexList[j].value ][0], colLen[i]);

    if ( isVarCol (i) ) { /* This column is variable-length, use corresponding VarCol instance */
     varCols[i].get (*state.varColStreams[i], atoi ( rowColumns [ curLen + state.sortedIndexList[j].value ].c_str() ), rowColumns [ curLen + state.sortedIndexList[j].value ]);
    }

    j++; 
    i++;
   }
   else { /* if it is not: seek without taking input */
    state.data.seekg (colLen[i], ios_base::cur);
    i++;
   }
  }

  pages[ state.page ].detachReader();
 }
 
 /* clear eof() and seek back one character so the stream would remain usable */
 if ( state.data.peek() == EOF && state.page == numPages - 1) {
  return false;
 }
 else {
  return true;
 }
}

bool Table::putRow (TableState& state, vector<string>& rowColumns, bool block) {
 string varPos;
 int tmpPos, varPosInt; 

 if (state.data.tellp () + rowLength >= MAX_PAGE_SIZE) {
  if (state.data.page == numPages - 1) { /* we have filled all the pages we have */
   if (pageAddMutex.lock ( false )) { /* we don't want more than one thread to increase the number of 
                                        pages */
                                           
    addPage ();
  
    pageAddMutex.unlock ();
   }
   else {
    pageAddCond.wait (); /* some other thread is already adding a page, wait for her */
   }
  }

  state.page += 1;
  state.data.close ();
  if ( !open ( state ) ) { return false; }
 }

 int rowNum = floor (state.data.tellp() / rowLength);
 bool locked = pages[ state.page ].writeLock (rowNum, block);

 if (!locked) { /* if we are non-blocking (i.e inserting) and writeLock returns false tell the caller 
                   so it would advance to the next row and try again */
  return false; 
 }

 if (state.columnNames.size() == 0) { /* write row in table's own order */
  int i = 0;

  if (numColumns != rowColumns.size()) {
   return false;
  }

  state.data.fill  ('\0');
  state.data.flags (ios::left);

  int rowNum = state.data.tellp() / rowLength; /* this must not return a float */
  pages[ state.page ].writeLock ( rowNum );

  while (i < numColumns) {
   state.data.width ( colLen[i] );

   if ( isVarCol (i) ) { /* Variable-length column */
    int prePos = state.data.tellg ();
    state.data.seekg (state.data.tellp(), ios_base::beg); /* synchornize get and put */

    state.data >> varPos;

    if (varPos.size () == 0) { /* we are inserting a new column seek to end of varcol stream */
     varPosInt = -1;
    }
    else {
     varPosInt = atoi (varPos.c_str());
    }

    tmpPos = varCols[i].put (*state.varColStreams[i], varPosInt, rowColumns[i]);

    if (tmpPos != varPosInt) { /* VarCol relocation took place */
     state.data.clear ();
     if (varPosInt != -1) state.data.seekp (- colLen[i], ios_base::cur);
     state.data << tmpPos;
    }

    state.data.seekg (prePos); /* put get pointer in its original place */
   }
   else {
    if (colLen[i] < rowColumns[i].size()) {
     rowColumns[i].erase (colLen[i]);
    }
 
    state.data << rowColumns[i];
   }
  
   i++;
  }

 }
 else { /* write row in a custom order provided by state */
  int len = state.sortedIndexList.size();

  if (rowColumns.size() != len) {
   return false;
  }

  int i = 0;
  int j = 0;
 
  state.data.fill ('\0');
  state.data.flags(ios::left);
	
  int rowNum = state.data.tellp() / rowLength; /* this must not return a float */
  pages[ state.page ].writeLock ( rowNum );

  while (i < numColumns && j < len) {
   if (i == state.sortedIndexList[j].key) {
    state.data.width ( colLen[i] );

    if ( isVarCol (i) ) { /* Variable-length column */
     int prePos = state.data.tellg ();
     state.data.seekg (state.data.tellp(), ios_base::beg); /* synchornize get and put */

     state.data >> varPos;

     varPosInt = atoi (varPos.c_str());
     tmpPos = varCols[i].put (*state.varColStreams[i], varPosInt, rowColumns[ state.sortedIndexList[j].value ]);

     if (tmpPos != varPosInt) { /* VarCol relocation took place */
      state.data.seekp (- colLen[i], ios_base::cur);
      state.data << tmpPos;
     }

     state.data.seekg (prePos); /* put get pointer in its original place */
    }
    else {
     if (colLen[i] < rowColumns[ state.sortedIndexList[j].value ].size()) {
      rowColumns[ state.sortedIndexList[j].value ].erase (colLen[i]);
     }
   
     state.data << rowColumns[ state.sortedIndexList[j].value ];

     j++; 
     i++;
    }
   }
   else {
    state.data.seekp (colLen[i], ios_base::cur);
    i++;
   }

  }

 }

 pages[ state.page ].writeUnlock ( rowNum );
 return true;
}

void Table::commitFormatModification () {
 formatModMutex.lock ();

 fstream new_data;  
 TableState state (*this);

 new_data.fill  ('\0');
 new_data.flags (ios_base::left);

 int i; /* loop over pages to writeLock */ 
 for (i = 0; i < numPages; i++) {
  pages[i].writeLock ();
 }

 /* actual data transfer starts here */
 TablePage tempPage; /* used for temporary page file name composition */

 vector<string> rowData;

 int rowCount = 0;
 int pageCount = 0; /* count of pages for new format */

 this->open (state);

 tempPage.set (tableName, 1);
 tempPage.sourceName += ".new";

 new_data.open (tempPage.sourceName.c_str(), fstream::out | fstream::trunc);

 while ( getRow(state, rowData) ) { /* getRow will automatically open the next page */

  if (rowCount * rowLengthMod > MAX_PAGE_SIZE) {
   new_data.close ();
   tempPage.set (tableName, ++pageCount);
   tempPage.sourceName += ".new";
   new_data.open (tempPage.sourceName.c_str(), fstream::out | fstream::trunc);
  }

  int j; /* loop over new columns */
  for (j = 0; j < numColumnsMod; j++) {
   new_data.width ( colLenMod[i] );

   if (j < numColumns) { /* we passed the number of columns in the old table, this column is new */
    new_data << '\0';
   }
   else { /* we may need to grab column data from old table */
    if (removedColumns[j]) { /* this column was in the old table, but it was removed from new format */
     new_data << '\0';
    }
    else { /* this column was in old table, and it is still needed in the new format */
     if (colLenMod[i] < rowData[ j ].size() ) {
      rowData[ j ].erase (colLenMod[i]);
     }
     new_data << rowData[j];
    }
   }
  }

  rowCount++;
 }

 state.data.close ();

 /* locking of reads should be done just before remove()              */
 /* but we do it before copying format to live containers in order to */
 /* avoid having to synchronize access to live-format containers      */

 /* loop over pages to readLock them */
 for (i = 0; i < numPages; i++) {
  pages[i].readLock ();
 }

 /* update live format containers */
 colName = colNameMod;
 colLen = colLenMod;
 colType = colTypeMod;
 colNameIndex = colNameIndexMod;
 rowLength = rowLengthMod;
 numColumns = numColumnsMod;
 varCols = varColsMod;

 int oldNumPages = numPages;
 numPages = pageCount; /* we have to update this within readLock */

 if (numPages < oldNumPages) { /* Remove excess pages from the pages vector */
  TablePage page;
  for (k = oldNumPages - 1; k >= numPages; k--) {
   remove ( pages[k].sourceName.c_str() );
   pages.pop_back ();
  }
 }
 else if (numPages > oldNumPages) { /* Add new pages to the pages vector */
  TablePage page;
  for (k = oldNumPages - 1; k < numPages; k++) {
   page.set (tableName, k);
   pages.push_back (page);
  }
 }

 numPagesModified = false; /* Do that within the readLock */

 string newName; /* loop over remaining pages to rename files */
 for (i = 0; i < numPages; i++) {
  remove ( pages[i].sourceName.c_str() );
  newName = pages[i].sourceName + ".new";
  rename ( newName.c_str(), pages[i].sourceName.c_str() );
  pages[i].readUnlock ();
  pages[i].writeUnlock ();
 }

 formatModified = false;
 formatModMutex.unlock ();

 saveFormat(); /* write down the new format to the format file */
}

Table::~Table () {
 if (formatModified) {
  saveFormat();
 }
 if (numPagesModified) {
  saveNumPages();
 }
}
